package com.google.code.motoo.sample.press;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;
import java.nio.charset.CharsetEncoder;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Random;

import com.google.code.motoo.core.ClientConnectionClosedException;

public class ConcurrentClient implements Runnable{
	private String host;
	private int port;
	private Selector selector = null; 
	private byte[] delimit = "\r\n".getBytes();
	private CharsetEncoder encoder = Charset.forName("GBK").newEncoder();
	private CharsetDecoder decoder = Charset.forName("GBK").newDecoder();
	private ByteBuffer in = ByteBuffer.allocate(100);
	private volatile boolean running = true;
	private List<SocketChannel> containor = new ArrayList<SocketChannel>();

	public ConcurrentClient(String host, int port) throws IOException {
		this.host = host;
		this.port = port;
		selector = Selector.open(); 
	}
	
	public void openChannel(int size) throws Exception {
		for (int i = 0; i < size; i++) {
			SocketChannel channel = SocketChannel.open(new InetSocketAddress(host, port));
			containor.add(channel);
			register(channel);
			Thread.sleep(100);
		}
	}
	
	public void register(SocketChannel channel) throws IOException {
		channel.configureBlocking(false);
		channel.register(selector, SelectionKey.OP_READ, ByteBuffer.allocate(400));
	}
	
	public void write(SocketChannel channel, Object message) throws IOException {
//		channel.configureBlocking(false);
//		channel.register(selector, SelectionKey.OP_WRITE, message);
		
		ByteBuffer block = encoder.encode(CharBuffer.wrap(message.toString()));
		ByteBuffer result = ByteBuffer.allocate(block.capacity() + delimit.length);
		result.put(block);
		result.put(delimit);
		result.flip();
		channel.write(result);
	}
	
	public void run() {
		while (running) {
			SelectionKey key = null;
			try {
				selector.select();
				Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
				while (iter.hasNext()) {
					key = iter.next();
					iter.remove();
					if (key.isValid()) {
						if (key.isReadable())
							doRead(key);
						else if (key.isWritable())
							doWrite(key);
					}
					
					key = null;
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}
	
	private void doRead(SelectionKey key) {
		SocketChannel channel = (SocketChannel) key.channel();
		ByteBuffer buffer = (ByteBuffer) key.attachment();
		int count;
		boolean match = false;
		try {
			count = channel.read(in);
			if (count > 0) {
				in.flip();
				while (in.hasRemaining()) {
					byte b = in.get();
					if (b == delimit[0]) {
						int position = in.position();
						match = true;
						for (int i = 1; i < delimit.length; i++) {
							if (delimit[i] != in.get()) {
								match = false;
								break;
							}
						}
						if (!match) {
							in.position(position);
						}
					} 
					
					if (match) {
						buffer.flip();
						CharBuffer charBuffer = decoder.decode(buffer);
						String content = charBuffer.toString();
						buffer.clear();
						if (content != null && !content.trim().equals(""))
						System.out.println(content);
					} else {
						buffer.put(b);
					}
				}
			}
			if (count < 0) {
				throw new ClientConnectionClosedException("client is closed!");
			}
			in.clear();
		} catch (Exception e) {
			e.printStackTrace();
		} 
		
	}

	private void doWrite(SelectionKey key) throws IOException {
		SocketChannel channel = (SocketChannel) key.channel();
		String message = (String) key.attachment();
		
		ByteBuffer block = encoder.encode(CharBuffer.wrap(message));
		ByteBuffer result = ByteBuffer.allocate(block.capacity() + delimit.length);
		result.put(block);
		result.put(delimit);
		channel.write(result);
	}
	
	public List<SocketChannel> getContainor() {
		return containor;
	}
	
	public static void main(String[] args) throws Exception {
		int size = 100;
		int rate = 10;
		int para = args.length;
		if (para > 0)
			if (args[0] != null)
				size = Integer.parseInt(args[0]);
		if (para > 1)
			if (args[1] != null)
				rate = Integer.parseInt(args[1]);
		Random random = new Random();
		ConcurrentClient client = new ConcurrentClient("192.168.1.210", 9000);
		client.openChannel(size);
		List<SocketChannel> list = client.getContainor();
		
		Thread thread = new Thread(client);
		thread.start();
		
		int count = (size * rate) / 100;
		while (true) {
			for (int i = 0; i < count; i++) {
				int index = random.nextInt(list.size());
				SocketChannel channel = list.get(index);
				client.write(channel, "hello");
			}
			Thread.sleep(1000);
		}
	}

}
